package com.gzyj.flink.broadcast;

import com.gzyj.flink.pojo.gps.MonDaoludengjiSet;
import com.gzyj.flink.pojo.gps.VehGps;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author lhj
 */
public class DaoludengjiBroadcastFun extends BroadcastProcessFunction<VehGps, MonDaoludengjiSet, VehGps> {

    MapStateDescriptor<String, MonDaoludengjiSet> monDaoludengjiSetMapStateDescriptor;

    public DaoludengjiBroadcastFun(MapStateDescriptor<String, MonDaoludengjiSet> monDaoludengjiSetMapStateDescriptor) {
        this.monDaoludengjiSetMapStateDescriptor = monDaoludengjiSetMapStateDescriptor;
    }


    @Override
    public void processElement(VehGps vehGps, BroadcastProcessFunction<VehGps, MonDaoludengjiSet, VehGps>.ReadOnlyContext readOnlyContext, Collector<VehGps> collector) throws Exception {
        ReadOnlyBroadcastState<String, MonDaoludengjiSet> readOnlyBcState = readOnlyContext.getBroadcastState(monDaoludengjiSetMapStateDescriptor);
        if (readOnlyBcState.contains(vehGps.getVehicleId())) {
            //存在车辆配置

            MonDaoludengjiSet monDaoludengjiSet = readOnlyBcState.get(vehGps.getVehicleId());

            int maxSpeed = 120;

            switch (vehGps.getRoadlevel()) {
                case "高速公路":
                    maxSpeed = monDaoludengjiSet.getGaosu();
                    break;
                case "国道":
                    maxSpeed = monDaoludengjiSet.getGuodao();
                    break;
                case "省道":
                    maxSpeed = monDaoludengjiSet.getShengdao();
                    break;
                case "县道":
                    maxSpeed = monDaoludengjiSet.getXiandao();
                    break;
                case "乡道":
                    maxSpeed = monDaoludengjiSet.getXiangdao();
                    break;
                default:
                    maxSpeed = monDaoludengjiSet.getQita();
                    break;
            }


            vehGps.setCalcroadlimited(maxSpeed);


            collector.collect(vehGps);


        } else {


            int maxSpeed = 120;

            switch (vehGps.getRoadlevel()) {
                case "高速公路":
                    switch (vehGps.getShiyongxingzhi()) {
                        case "道路旅客运输":
                            maxSpeed = 120;
                            break;
                        case "网络预约出租车":
                            maxSpeed = 120;
                            break;
                        case "出租车":
                            maxSpeed = 120;
                            break;
                        case "道路危险货物运输":
                            maxSpeed = 80;
                            break;
                        case "道路货物运输":
                            maxSpeed = 120;
                            break;
                        case "班线客运客车":
                            maxSpeed = 120;
                            break;
                        default:
                            maxSpeed = 120;
                            break;
                    }
                    break;

                case "国道":
                    switch (vehGps.getShiyongxingzhi()) {
                        case "道路旅客运输":
                            maxSpeed = 80;
                            break;
                        case "网络预约出租车":
                            maxSpeed = 80;
                            break;
                        case "出租车":
                            maxSpeed = 80;
                            break;
                        case "道路危险货物运输":
                            maxSpeed = 80;
                            break;
                        case "道路货物运输":
                            maxSpeed = 80;
                            break;
                        case "班线客运客车":
                            maxSpeed = 80;
                            break;
                        default:
                            maxSpeed = 80;
                            break;
                    }
                    break;
                case "省道":
                    switch (vehGps.getShiyongxingzhi()) {
                        case "道路旅客运输":
                            maxSpeed = 80;
                            break;
                        case "网络预约出租车":
                            maxSpeed = 80;
                            break;
                        case "出租车":
                            maxSpeed = 80;
                            break;
                        case "道路危险货物运输":
                            maxSpeed = 80;
                            break;
                        case "道路货物运输":
                            maxSpeed = 80;
                            break;
                        case "班线客运客车":
                            maxSpeed = 80;
                            break;
                        default:
                            maxSpeed = 80;
                            break;
                    }
                    break;

                case "县道":
                    switch (vehGps.getShiyongxingzhi()) {
                        case "道路旅客运输":
                            maxSpeed = 60;
                            break;
                        case "网络预约出租车":
                            maxSpeed = 60;
                            break;
                        case "出租车":
                            maxSpeed = 60;
                            break;
                        case "道路危险货物运输":
                            maxSpeed = 60;
                            break;
                        case "道路货物运输":
                            maxSpeed = 60;
                            break;
                        case "班线客运客车":
                            maxSpeed = 60;
                            break;
                        default:
                            maxSpeed = 60;
                            break;
                    }
                    break;

                case "乡道":
                    switch (vehGps.getShiyongxingzhi()) {
                        case "道路旅客运输":
                            maxSpeed = 40;
                            break;
                        case "网络预约出租车":
                            maxSpeed = 40;
                            break;
                        case "出租车":
                            maxSpeed = 40;
                            break;
                        case "道路危险货物运输":
                            maxSpeed = 40;
                            break;
                        case "道路货物运输":
                            maxSpeed = 40;
                            break;
                        case "班线客运客车":
                            maxSpeed = 40;
                            break;
                        default:
                            maxSpeed = 40;
                            break;
                    }
                    break;

                default:
                    switch (vehGps.getShiyongxingzhi()) {
                        case "道路旅客运输":
                            maxSpeed = 60;
                            break;
                        case "网络预约出租车":
                            maxSpeed = 60;
                            break;
                        case "出租车":
                            maxSpeed = 60;
                            break;
                        case "道路危险货物运输":
                            maxSpeed = 60;
                            break;
                        case "道路货物运输":
                            maxSpeed = 60;
                            break;
                        case "班线客运客车":
                            maxSpeed = 60;
                            break;
                        default:
                            maxSpeed = 60;
                            break;
                    }
                    break;
            }

            vehGps.setCalcroadlimited(maxSpeed);

            collector.collect(vehGps);

        }


    }

    @Override
    public void processBroadcastElement(MonDaoludengjiSet monDaoludengjiSet, BroadcastProcessFunction<VehGps, MonDaoludengjiSet, VehGps>.Context context, Collector<VehGps> collector) throws Exception {
        BroadcastState<String, MonDaoludengjiSet> bcState = context.getBroadcastState(monDaoludengjiSetMapStateDescriptor);
        bcState.put(monDaoludengjiSet.getVehid(), monDaoludengjiSet);
    }
}
